package com.gitee.wsl.ext.coroutines

import androidx.compose.runtime.snapshotFlow
import com.gitee.wsl.concurrent.api.SuspendRunnable
import com.gitee.wsl.concurrent.api.cancel.Cancellable
import com.gitee.wsl.coroutines.strategy.safely.ExceptionHandler
import com.gitee.wsl.coroutines.strategy.safely.defaultSafelyExceptionHandler
import com.gitee.wsl.coroutines.strategy.safely.safely
import com.gitee.wsl.ext.any.runIf
import com.gitee.wsl.flow.coroutines.subscribe
import com.gitee.wsl.flow.coroutines.subscribeAsync
import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.launch
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.cancellable
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.onStart
import kotlinx.coroutines.isActive
import kotlinx.coroutines.plus
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.yield
import timber.log.Timber
import kotlin.coroutines.CoroutineContext
import kotlin.coroutines.EmptyCoroutineContext
import kotlin.coroutines.cancellation.CancellationException
import kotlin.time.Duration

private object EmptyScope : CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = EmptyCoroutineContext
}


suspend inline fun CoroutineScope.activeCheck() = yield()

suspend inline fun CoroutineScope.activeRun(block: SuspendRunnable) {
    isActive.runIf(block)
}

suspend inline fun CoroutineScope.activeLoop(block: SuspendRunnable) {
   while(isActive){
       block.run()
   }
}


/**
 * Launch a new coroutine using [this] scope and [context],
 * but catch any [Throwables] using [onError] and execute [block].
 *The coroutines launched inside [block]'s scope will NOT cancel the scope.
 */
fun CoroutineScope.launchCatching(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    onError: CoroutineContext.(Throwable) -> Unit,
    block: suspend CoroutineScope.() -> Unit,
): Job {
    val handler = CoroutineExceptionHandler { it, e ->
        onError(it, e)
    }
    val scope = this + handler + SupervisorJob()
    return scope.launch(context, start, block)
}

/**
 * Executes the [block] first, then executes the [delay], and then repeats.
 */
suspend fun CoroutineScope.preRepeat(interval: Duration, block: suspend CoroutineScope.() -> Unit) {
    while (true) {
        block(this)
        delay(interval)
    }
}

/**
 * Executes the [delay] first, then executes the [block], and then repeats.
 */
suspend fun CoroutineScope.postRepeat(interval: Duration, block: suspend CoroutineScope.() -> Unit) {
    while (true) {
        delay(interval)
        block(this)
    }
}

fun <T> CoroutineScope.subscribe(flow: Flow<T>, body: suspend (T) -> Unit): Job =
    flow.cancellable().onEach { body(it) }.launchIn(this)

fun CoroutineScope.safeLaunch(
    exceptionHandler: CoroutineExceptionHandler = defaultExceptionHandler,
    block: suspend CoroutineScope.() -> Unit,
): Job = this.launch(
    context = exceptionHandler,
    block =  block
)

fun CoroutineScope.launch(
    onError: (throwable: Throwable) -> Unit = { _ -> },
    block: suspend CoroutineScope.() -> Unit,
): Job = launch(
    context = CoroutineExceptionHandler { _, throwable ->
        Timber.e("CoroutineException", throwable.message, throwable)
        onError(throwable)
    },
    block =block
)

fun CoroutineScope.countDownFlow(
    time: Int = 60,
    onStart: (suspend () -> Unit)? = null,
    onEach: (suspend (Int) -> Unit)? = null,
    onCancel: (suspend (msg: String?) -> Unit)? = null,
    onCompletion: (suspend () -> Unit)? = null
): Job {
    return (time downTo 0)
        .asFlow()
        .cancellable()
        .flowOn(Dispatchers.Default)
        .onStart {
            onStart?.invoke()
        }
        .onEach {
            onEach?.invoke(it)
            delay(1000L)
        }
        .onCompletion {
            if (it == null) {
                onCompletion?.invoke() // 正常完成
            } else {
                onCancel?.invoke(it.message) // 取消时执行取消回调
            }
        }
        .launchIn(this)

}

internal fun <T> CoroutineScope.async(start: CoroutineStart, callback: suspend () -> T): Deferred<T> = async(coroutineContext, start = start) {
    try {
        callback()
    } catch (e: Throwable) {
        if (e is CancellationException) throw e
        e.printStackTrace()
        throw e
    }
}

fun<T> CoroutineScope.collectToSnapShotFlow(
    linesParameters: T,
    makeUpdateData: (T) -> Unit
) {
    this.launch {
        snapshotFlow {
            linesParameters
        }.collect {
            makeUpdateData(it)
        }
    }
}


/**
 * Executes the given [block] with [async] function from the [CoroutineScope] with maximum [parallelism].
 *
 * The [block] is executed concurrently for each element in the provided scope with a maximum of [parallelism] coroutines.
 *
 * @param T the type of the elements in the collection
 * @param collection the collection to iterate over
 * @param parallelism the maximum number of coroutines to run concurrently
 * @param block the function to be executed concurrently
 * @return a list of [Deferred] that represent the coroutines running the [block]
 */
inline fun <T, R> CoroutineScope.parallel(
    parallelism: Int,
    collection: Iterable<T>,
    crossinline block: suspend (T) -> R
): List<Deferred<R>> {

    val semaphore = Semaphore(parallelism)
    return collection.map {
        this@parallel.async {
            semaphore.withPermit {
                block(it)
            }
        }
    }
}


fun <T> CoroutineScope.actor(
    capacity: Int = Channel.UNLIMITED,
    block: suspend (T,channel:Channel<T>) -> Unit
): Channel<T> {
    val channel = Channel<T>(capacity)
    channel.consumeAsFlow().subscribe(this){
        block(it,channel)
    }
    return channel
}

inline fun <T> CoroutineScope.safeActor(
    capacity: Int = Channel.UNLIMITED,
    noinline onException: ExceptionHandler<Unit> = defaultSafelyExceptionHandler,
    crossinline block: suspend (T,channel:Channel<T>) -> Unit
): Channel<T> = actor(
    capacity
) {it,channel ->
    safely(onException) {
        block(it,channel)
    }
}


fun <T> CoroutineScope.actorAsync(
    channelCapacity: Int = Channel.UNLIMITED,
    markerFactory: suspend (T) -> Any? = { null },
    block: suspend (T) -> Unit
): Channel<T> {
    val channel = Channel<T>(channelCapacity)
    channel.consumeAsFlow().subscribeAsync(this, markerFactory, block)
    return channel
}

inline fun <T> CoroutineScope.safeActorAsync(
    channelCapacity: Int = Channel.UNLIMITED,
    noinline onException: ExceptionHandler<Unit> = defaultSafelyExceptionHandler,
    noinline markerFactory: suspend (T) -> Any? = { null },
    crossinline block: suspend (T) -> Unit
): Channel<T> = actorAsync(
    channelCapacity,
    markerFactory
) {
    safely(onException) {
        block(it)
    }
}


fun CoroutineScope.loopCatching(
    handler: (Throwable) -> Unit,
    block: suspend CoroutineScope.() -> Unit,
): Job = launch {
    while (isActive) {
        try {
            block()
        } catch (t: Throwable) {
            handler(t)
        }
        delay(10)
    }
}

 fun <T> CoroutineScope.lazyAsync(
    context: CoroutineContext = EmptyCoroutineContext,
    block: suspend CoroutineScope.() -> T,
): Deferred<T> = async(
    context = context,
    start = CoroutineStart.LAZY,
    block = block,
)


/**
 * 注册一个函数，用于在[CoroutineScope]被取消时执行
 *
 * @param block
 * @receiver
 */
inline fun CoroutineScope.onCancellation(
    crossinline block: () -> Unit
) = launch {
    try {
        awaitCancellation()
    } finally {
        block()
    }
}




/**
 * Description:
 * @author Junerver
 * date: 2024/1/31-11:16
 * Email: junerver@gmail.com
 * Version: v1.0
 */
/**
 * 一个可取消的重复间隔执行方法
 */
fun CoroutineScope.launchPeriodicAsync(
    repeatMillis: Long,
    action: () -> Unit
) = this.async {
    if (repeatMillis > 0) {
        while (isActive) {
            action()
            delay(repeatMillis)
        }
    } else {
        action()
    }
}